package yz.mr.FiltrationText2;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

/**
 * 对于各班级中的学生总分进行排序，要求取出各班级地前三名
 *
 * 对于需要做数据排序的操作，可以使用java里面的方式排序，同时Reduce端写出的数据类型可以使用自定义的数据类型
 *
 * 分析
 *      需要读取ReduceJoin的结果数据，之后在对数据进行处理，将相同班级的学生数据汇聚在一个Reduce中进行处理，
 *      需要取出对应的学生信息，并包装成一个学生对象，并将所有的学生对象存储在一个List集合中，对应List集合的数据在做数据排序，对应取出下标的数据
 *
 * Mapper阶段：
 *      ①读取数据并且进行切分，读出数据信息
 *      ②对班级信息做为key，将其他的信息作为value
 *
 * Reduce阶段：
 *      将相同班级的学生数据汇聚在一个Reduce中进行处理，
 *      需要取出对应的学生信息，并包装成一个学生对象，并将所有的学生对象存储在一个List集合中，
 *      对应List集合的数据在做数据排序，对应取出下标的数据
 *
 */
public class FiltrationText2 {
    public static void main(String[] args) throws Exception{

        //1.创建Job操作对象
        //2.创建配置类对象
        Configuration entries = new Configuration();
        Job job = Job.getInstance(entries);

        //3.配置job
        job.setJobName("FiltrationText2");
        job.setJarByClass(FiltrationText2.class);

        //4.设置Mapper以及输出类型
        job.setMapperClass(FiltrationText2Mapper.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);  //这里的Mapper和Reduce类型要一致

        job.setReducerClass(FiltrationText2Reduce.class);
        //6.设置最终地输出数据类型
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);

        //7.设置输入输出路径(这是本地的路径)  输入路径既可以指定路径也可以指定目录
        FileInputFormat.addInputPath(job,new Path("output/ReduceCount"));
        FileOutputFormat.setOutputPath(job,new Path("output/Filtration2Count"));

        //8.提交执行当前JOB
        job.waitForCompletion(true);

    }
}
